package com.shujia.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo04MySource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 使用自定义source
    val myDS: DataStream[Int] = env.addSource(new MySource())
    myDS.print()

    env.execute()

  }

}

class MySource extends SourceFunction[Int] {
  /**
   * 每次任务启动时会执行一次
   * ctx：Source的上下文对象 用于将数据发送给Flink任务
   *
   */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
    // 每调用一次collect方法 即可获得一条数据
    var i: Int = 0
    while (true) {
      ctx.collect(i)
      i += 1
      Thread.sleep(1000)
    }
  }

  // 当程序结束时会被调用 主要用于回收资源
  override def cancel(): Unit = {

  }
}
